/*
 * Create roles if they do not yet exist. Roles are server-wide objects that
 * are not owned by the extension, so they could easily exist if the extension
 * was previously created and dropped, or exists in multiple databases.
 */
DO LANGUAGE plpgsql $$
BEGIN
	IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'lake_read') THEN
		CREATE ROLE lake_read;
	END IF;
	IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'lake_write') THEN
		CREATE ROLE lake_write;
	END IF;
	IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'lake_read_write') THEN
		CREATE ROLE lake_read_write;
	END IF;
	GRANT lake_read TO lake_read_write;
	GRANT lake_write TO lake_read_write;
END;
$$;

-- we need to create a schema to hold the internal functions
-- that are used by the pg_lake_table extension
CREATE SCHEMA __lake__internal__nsp__;
GRANT USAGE ON SCHEMA __lake__internal__nsp__ TO public;

/* New composite types will be created here, so we want to allow any user to
 * create on this schema. */
CREATE SCHEMA lake_struct;
GRANT USAGE ON SCHEMA lake_struct TO public;
GRANT CREATE ON SCHEMA lake_struct TO public;

/* this schema contains user-facing utiltiy functions which requires write access */
CREATE SCHEMA lake_engine;
GRANT USAGE ON SCHEMA lake_engine TO lake_write;

-- __lake_read_table is a placeholder for the file read functions
-- in DuckDB (e.g. read_parquet).
--
-- We'll never call this function on Postgres, but only pushdown
-- the function to the query engine.
CREATE OR REPLACE FUNCTION __lake__internal__nsp__.__lake_read_table(table_name TEXT, unique_table_id INT)
RETURNS SETOF record AS $$
BEGIN
  RETURN;
END;
$$ LANGUAGE plpgsql;

-- skeletons for the time_bucket function to be used in the deparser
CREATE FUNCTION __lake__internal__nsp__.time_bucket(interval, timestamp, timestamp)
 RETURNS timestamp
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.time_bucket(interval, timestamptz, timestamptz)
 RETURNS timestamptz
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;


/*
 * to_date is meant to deal with the common representation of dates in Parquet (days
 * since epoch stored in int32). We use double precision for convenience and to match
 * the signature of to_timestamp(double precision).
 */
CREATE OR REPLACE FUNCTION pg_catalog.to_date(double precision)
 RETURNS date
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_to_date$function$;
COMMENT ON FUNCTION to_date(double precision)
IS 'convert days since UNIX epoch to date';

-- skeletons for the substring function to be used in the deparser
CREATE FUNCTION __lake__internal__nsp__.substring_pg(text, int)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.substring_pg(text, int, int)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.nullify_any_type(record)
 RETURNS record
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

-- emit a weird name for now() in the deparser to do search & replace
CREATE FUNCTION __lake__internal__nsp__.__lake_now()
 RETURNS timestamptz
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.map_extract("any","any")
 RETURNS "any"
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

/* 
 * Whenever a file is entered to the deletion queue, it'll be 
 * removed in lake_engine.orphaned_file_retention_period.
 */
CREATE TABLE lake_engine.deletion_queue (
    path text PRIMARY KEY,
    table_name regclass,
    orphaned_at timestamptz,
	retry_count int DEFAULT 0,
	is_prefix bool DEFAULT false
);
GRANT SELECT ON lake_engine.deletion_queue TO lake_write;

CREATE OR REPLACE FUNCTION lake_engine.flush_deletion_queue(table_name regclass)
 RETURNS SETOF text
 LANGUAGE C
 STRICT
AS 'MODULE_PATHNAME', $function$flush_deletion_queue$function$;
REVOKE ALL ON FUNCTION lake_engine.flush_deletion_queue(table_name regclass) FROM public;
GRANT EXECUTE ON FUNCTION lake_engine.flush_deletion_queue(table_name regclass) TO lake_write;

CREATE SEQUENCE lake_engine.operationid_seq;
GRANT USAGE ON lake_engine.operationid_seq TO lake_write;

/* 
 * Files that are generated by in-progress transactions
 * are recorded here. If a transaction is successfully
 * committed, its in-progress tables are removed by the
 * same transaction. If not commited, the files are removed
 * from remote storage and this table on VACUUM.
 */
CREATE TABLE lake_engine.in_progress_files (
    path text PRIMARY KEY,
    operation_id int8,
    is_prefix bool
);
GRANT SELECT ON lake_engine.in_progress_files TO lake_write;

CREATE OR REPLACE FUNCTION lake_engine.flush_in_progress_queue()
 RETURNS SETOF text
 LANGUAGE C
 STRICT
AS 'MODULE_PATHNAME', $function$flush_in_progress_queue$function$;
REVOKE ALL ON FUNCTION lake_engine.flush_in_progress_queue() FROM public;
GRANT EXECUTE ON FUNCTION lake_engine.flush_in_progress_queue() TO lake_write;

/*
 * Function that can be used to break pushdown and pull data into PostgreSQL.
 *
 * Returning the input is useful, because then it can be added to any expression.
 */
CREATE FUNCTION pg_catalog.to_postgres(val "any")
 RETURNS "any"
 LANGUAGE c
 STRICT STABLE PARALLEL SAFE
AS 'MODULE_PATHNAME', $function$pg_lake_to_postgres$function$;
COMMENT ON FUNCTION pg_catalog.to_postgres("any") IS 'force pulling data into postgres';


/*
 * We need to push down `regexp_like` and rename to `regexp_matches`;
 * unfortunately postgres also has `regexp_matches` but it has the wrong
 * signature, so we need stubs here to lookup as our replacement.
 */

CREATE FUNCTION __lake__internal__nsp__.lake_regexp_matches(text,text,bool)
 RETURNS bool
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.xor(int2,int2)
 RETURNS int2
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.xor(int4,int4)
 RETURNS int4
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.xor(int8,int8)
 RETURNS int8
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

/*
 * DuckDB name for div
 */
CREATE FUNCTION __lake__internal__nsp__.fdiv(numeric, numeric)
 RETURNS numeric
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

/*
 * DuckDB name for mod
 */
CREATE FUNCTION __lake__internal__nsp__.fmod(numeric, numeric)
 RETURNS numeric
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

---------------------------

/*
 * DuckDB integer division
 */
CREATE FUNCTION __lake__internal__nsp__.divide(int2, int2)
 RETURNS int2
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int2, int4)
 RETURNS int2
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int2, int8)
 RETURNS int2
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int4, int2)
 RETURNS int4
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int4, int4)
 RETURNS int4
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int4, int8)
 RETURNS int4
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int8, int2)
 RETURNS int8
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int8, int4)
 RETURNS int8
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.divide(int8, int8)
 RETURNS int8
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__."trim"(text)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__."trim"(text, text)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.strftime(timestamp, text)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.strftime(timestamptz, text)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.printf(text, variadic "any")
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.lake_nth_suffix(int)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.json_array_length(jsonb)
 RETURNS int
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

-- create wrapper jsonb() function for all the types
-- that can be casted to jsonb. This function corresponds
-- to the macro with the same name in pgduck_server/Duckdb
CREATE FUNCTION __lake__internal__nsp__.jsonb(jsonb)
 RETURNS jsonb
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.jsonb(text)
 RETURNS jsonb
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.jsonb(json)
 RETURNS jsonb
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.jsonb(cstring)
 RETURNS jsonb
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.jsonb(varchar)
 RETURNS jsonb
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.jsonb(bpchar)
 RETURNS jsonb
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

-- skeletons for the substring function to be used in the deparser
CREATE FUNCTION __lake__internal__nsp__.generate_series_int(int, int)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;

CREATE FUNCTION __lake__internal__nsp__.generate_series_int_step(int, int, int)
 RETURNS text
 LANGUAGE C
 IMMUTABLE PARALLEL SAFE STRICT
AS 'MODULE_PATHNAME', $function$pg_lake_internal_dummy_function$function$;
